package com.easou.ps.master.query.qs;

import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

import com.easou.ps.common.servicehandle.Call;
import com.easou.ps.common.servicehandle.ConnectionGuard;
import com.easou.ps.common.servicehandle.ServiceAddress;
import com.easou.ps.master.kit.ConfigFactory;
import com.easou.ps.thrift.qs.QSRPC.call_args;
import com.easou.ps.thrift.qs.QSRPC.call_result;
import com.easou.ps.thrift.qs.QSRequest;
import com.easou.ps.thrift.qs.QSResult;

public class QSProxy {
	
	private static final Logger LOG = Logger.getLogger("query");
	private static final int QS_TIMEOUT = ConfigFactory.getSysConfig().getInt("rpc.qs.timeout", 100);

	private final ServiceAddress address;
	
	AtomicInteger idGenerator = new AtomicInteger(0);
	
	public QSProxy(ServiceAddress address) {
		this.address = address;
	}
	
	public QSResult call(QSRequest query) {
		int seqid = idGenerator.getAndIncrement();
		String func = "call";
		
		call_args args = new call_args();
		args.setReq(query);
		
		call_result result = new call_result();

		Call call = new Call(seqid, func, args, result);
		
		QSResult ret = null;
		ConnectionGuard conn = address.getConnection(seqid);
		if (conn == null) {
			LOG.error("QS not running:" + address);
			return ret;
		}
		if (!conn.sendCall(call)) {
			LOG.error("[qs] send request to qs fail, qs[" + address + "], req:" + query);
		} else {
			call.waitComplete(QS_TIMEOUT);
			if (call.isFail()) {
				LOG.error("[qs] error occured when waiting qs result, qs[" + address + "], req:" + query);
			} else if (call.isTimeout()) {
				LOG.error("[qs] timeout when waiting qs result, qs[" + address + "], req:" + query);
			} else {
				ret = ((call_result)call.getResult()).success;
			}
		}
		return ret;
	}
	
	@Override
	public String toString(){
		return this.address.toString();
	}
}
